package com.yujiahao.bigdata.sql

import org.apache.spark.sql.expressions.{Aggregator, MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql._
import org.apache.spark.sql.types.{DataType, LongType, StructField, StructType}

object SparkSQL01_UDAF_Old {
  def main(args: Array[String]): Unit = {
    //TODO 1、SparkSQL直接构建无法构建，包是私有的，使用伴生对象伴生类构建builder
    //SparkSQL对SparkCore进行了封装，所以环境也要发生变化

    val spark = SparkSession.builder
      .master("local")
      .appName("Word Count")
      .getOrCreate()
    //构建Dataframe
    val df: DataFrame = spark.read.json("data/user.json")

    //使用SQL的方式访问DataFrame

    df.createOrReplaceTempView("user")
    spark.sql("select name from user")
    //TODO 使用老的API进行函数的自定义：UDAF，这里在注册方式的声明上要注意，直接new自己定义的类
    //如果想在SQL语句中执行自定义的聚合函数，那么一般需要提前声明UDAF，老版本的就直接new就行
    spark.udf.register("myAvg", new MyAvgAge)
    spark.sql("select myAvg(age) from user").show()

    //TODO 2、关闭资源
    spark.stop()
  }

  //Todo 自定义聚合函数（UDAF）（弱类型）不考虑泛型
  // 1、继承类UserDefinedAggregateFunction
  // 2、实现8种方法

  class MyAvgAge extends UserDefinedAggregateFunction {

    //TODO 重写方法一：输入的数据结构，
    override def inputSchema: StructType = {
     //1.1、发现StructType是一个样例类，可以直接调用，里面需要一个Array数组,那就给他传入一个数组
      StructType(
       //1.2、点进去发现需要传递一个StructField类型的因此点击StructField查看
        Array(
          //1.3、点进去发现还是一个样例类，那么就直接调用其中的属性，意思是将什么样的数据类型传递到函数当中，结构是什么样子的
          //1.4、我们这里需要对年龄进行处理
          StructField("age",LongType)
        )
      )
    }
    //TODO 重写方法二：这里处理缓冲区的数据类型，我们参考方法一，只不过这里有两个类型
    override def bufferSchema: StructType = {
      StructType(Array(
        StructField("total",LongType),
        StructField("cnt",LongType)
      ))
    }
    //TODO 重写方法三：返回值类型--返回一个LongType类型的
    override def dataType: DataType = LongType

    //TODO 重写方法四：计算稳定性-就是说计算过程中的数据每次都是稳定的不会改变的
    override def deterministic: Boolean = true

    //TODO 重写方法五：缓冲区初始化,没有返回值，因为缓冲区就两个数，所以就直接将0分别更新到里面去完成初始化
    override def initialize(buffer: MutableAggregationBuffer): Unit = {
      buffer.update(0,0L)
      buffer.update(1,0L)
    }

    //TODO 重写方法六：将输入的数据更新到缓冲区，注意这里还是没有返回值，因此直接更新
    override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
      //注意缓冲区的第一个数值是年龄的总和，所以输入的值与原来的值相加
      buffer.update(0,buffer.getLong(0) + input.getLong(0))
      //但是第二个是计算年龄的个数，所以就要计算总的个数，进来一个加一个
      buffer.update(1,buffer.getLong(1) + 1)
    }

    //TODO 重写方法七： 合并缓冲区，这里是无返回值，参数一底层是有updata方法的因此只能更新这个
    override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
      buffer1.update(0,buffer1.getLong(0) + buffer2.getLong(0))
      buffer1.update(1,buffer1.getLong(1) + buffer2.getLong(1))
    }
    //TODO 重写方法八：计算数据，这里也无返回值就直接将缓冲区里面的数据计算就行
    override def evaluate(buffer: Row): Any = {
      //因为都是Row所以直接通过gerLong方法拿到两个数进行相除
      buffer.getLong(0) / buffer.getLong(1)
    }
  }

  //使用样例类将缓冲区里面的类型进行封装
  case class AvgVuffer(var total: Long, var cnt: Long)
}
